package com.ksyun.campus.client.util;


import com.ksyun.campus.client.constant.ZkConstant;
import com.ksyun.campus.client.domain.MetaServer;
import com.ksyun.campus.client.domain.ModeType;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.RetryNTimes;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
public class ZkUtil {
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            2,
            3,
            60,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(2),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );

    private String address = "10.0.0.201:2181";

    private int DEFAULT_SESSION_TIMEOUT_MS = 15000;

    private int DEFAULT_CONNECTION_TIMEOUT_MS = 15000;

    public static ConcurrentHashMap<String, MetaServer> metaServerStore = new ConcurrentHashMap<>();

    public ZkUtil() {
    }

    public ZkUtil(String address) {
        this.address = address;
    }

    /**
     * 开始监听服务，主线程需要等待 zk连接成功且监听之后才能操作
     */
    public void start() {
        threadPoolExecutor.execute(() -> {
            synchronized (this.address) {
                this.postCons();
                this.address.notifyAll();
            }
        });
        synchronized (this.address) {
            try {
                address.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private void postCons() {
        // todo 初始化，与zk建立连接，注册监听路径，当配置有变化随时更新
        log.info("zk监听开始启动~");
        CuratorFramework curatorFramework = curatorFramework();
        //给指定节点的子节点注册监听
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, ZkConstant.META_SERVER, true);
        pathChildrenCache.getListenable().addListener((client, event) -> {
            PathChildrenCacheEvent.Type type = event.getType();
            String path = event.getData().getPath();
            if (type == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                log.info("元数据服务{}被添加~", path);
                byte[] data = event.getData().getData();
                MetaServer metaServer = (MetaServer) ByteConvertUtil.toObj(data, MetaServer.class);
                metaServerStore.put(path, metaServer);
            }
            if (type == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                log.info("元数据服务{}被移除~", path);
                metaServerStore.remove(path);
            }
            if (type == PathChildrenCacheEvent.Type.CHILD_UPDATED) {
                log.info("元数据服务{}更新~", path);
                byte[] data = event.getData().getData();
                MetaServer metaServer = (MetaServer) ByteConvertUtil.toObj(data, MetaServer.class);
                metaServerStore.put(path, metaServer);
            }
        });
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private CuratorFramework curatorFramework() {
        CuratorFramework curatorFramework = null;
        try {
            curatorFramework = CuratorFrameworkFactory.newClient(
                    address,
                    DEFAULT_SESSION_TIMEOUT_MS,
                    DEFAULT_CONNECTION_TIMEOUT_MS,
                    new RetryNTimes(Integer.valueOf(2), Integer.valueOf(3000))
            );
            curatorFramework.start();
            //阻塞直到连接成功
            curatorFramework.blockUntilConnected();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return curatorFramework;
    }

    /**
     * 获取元数据的主节点
     * @return
     */
    public static MetaServer getMetaMaster(){
        Set<Map.Entry<String, MetaServer>> entries = ZkUtil.metaServerStore.entrySet();
        for (Map.Entry<String, MetaServer> mateServerEntry : entries){
            MetaServer res = mateServerEntry.getValue();
            if(res.getMode().equals(ModeType.Master)){
                return res;
            }
        }
        return null;
    }


    /**
     * 获取元数据的主节点
     * @return
     */
    public static MetaServer getMetaSlave(){
        Set<Map.Entry<String, MetaServer>> entries = ZkUtil.metaServerStore.entrySet();
        for (Map.Entry<String, MetaServer> mateServerEntry : entries){
            MetaServer res = mateServerEntry.getValue();
            if(res.getMode().equals(ModeType.Follower)){
                return res;
            }
        }
        return null;
    }

    /**
     * 获取所有的元数据节点
     */
    @Deprecated
    public static List<MetaServer> getMetaList(){
        Set<Map.Entry<String, MetaServer>> entries = ZkUtil.metaServerStore.entrySet();
        List<MetaServer> metaServers = entries.stream().map(Map.Entry::getValue).collect(Collectors.toList());
        return metaServers;
    }
}
